package com.bsoft.advance.client.proxy;

import com.bsoft.advance.client.common.RequestMetadata;
import com.bsoft.advance.client.transport.RpcTransportFactory;
import com.bsoft.advance.core.common.RpcClientProperties;
import com.bsoft.advance.core.common.RpcRequest;
import com.bsoft.advance.core.common.RpcResponse;
import com.bsoft.advance.core.common.ServiceInfo;
import com.bsoft.advance.core.protocol.MessageHeader;
import com.bsoft.advance.core.protocol.MessageProtocol;
import com.bsoft.advance.core.register.DiscoveryService;
import lombok.extern.slf4j.Slf4j;

import java.lang.reflect.Method;
import java.net.InetSocketAddress;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicInteger;

@Slf4j
public class RpcRequestHandler {
    /**
     * rpc请求公共处理类
     *
     * @param properties
     * @param discoveryService
     * @param clazz
     * @param method
     * @param args
     * @return
     * @throws Exception
     */
    public static Object invokeRequest(RpcClientProperties properties, DiscoveryService discoveryService, Class<?> clazz, Method method, Object[] args) throws Exception {
        // 从注册中心发现服务信息
        ServiceInfo serviceInfo = discoveryService.discovery(clazz.getName());
        if (serviceInfo == null) {
            throw new IllegalStateException("该服务未注册");
        }
        // 组装消息体
        RpcRequest message = new RpcRequest();
        message.setServiceName(clazz.getName());
        message.setMethodName(method.getName());
        message.setParameterTypes(method.getParameterTypes());
        message.setParameters(args);

        // 组装协议
        MessageProtocol<RpcRequest> protocol = new MessageProtocol<>();
        protocol.setBody(message);
        MessageHeader header = MessageHeader.build(properties.getSerializationType().name());
        protocol.setHeader(header);

        // 组装元数据
        RequestMetadata metadata = RequestMetadata.builder()
                .socketAddress(new InetSocketAddress(serviceInfo.getAddress(), serviceInfo.getPort()))
                .protocol(protocol)
                .timeout(properties.getTimeout())
                .build();

        // 客户端执行发送请求
        //MessageProtocol<RpcResponse> response = invokeRequest(metadata);
        // 失败转移策略执行发送请求
        MessageProtocol<RpcResponse> response = failOverInvokeRequest(properties,discoveryService, clazz.getName(), metadata);

        if (response.isSuccess()) {
            return response.getBody().getData();
        }else{
            throw new RuntimeException(response.getBody().getErrMsg());
        }
    }


    /**
     * 客户端发送请求
     */
    private static MessageProtocol<RpcResponse> invokeRequest(RequestMetadata metadata) throws Exception {

        MessageProtocol<RpcResponse> response = null;

        try {
            response = RpcTransportFactory.getDefaultTransportInstance().send(metadata);
        } catch (TimeoutException e) {
            log.error("RPC request timeout for {}s", metadata.getTimeout() / 1000);
            throw new RuntimeException("request timeout");
        }

        return response;
    }

    /**
     * 已经重试次数
     */
    private static AtomicInteger retryTimes = new AtomicInteger();

    /**
     * 客户端发送请求(集群容错)
     */
    private static MessageProtocol<RpcResponse> failOverInvokeRequest(RpcClientProperties properties, DiscoveryService discoveryService, String serviceName, RequestMetadata metadata) throws Exception {
        MessageProtocol<RpcResponse> response = null;
        if (retryTimes.get() > properties.getRetryTime()){
            throw new RuntimeException("request timeout [failover : max retry]");
        }
        try {
            // 原子自增
            retryTimes.getAndIncrement();
            response = RpcTransportFactory.getDefaultTransportInstance().send(metadata);
        } catch (TimeoutException e) {
            ServiceInfo otherServiceInfo = discoveryService.discovery(serviceName);
            if (otherServiceInfo == null) {
                throw new RuntimeException("request timeout [failover : no more available server]");
            }
            metadata.setSocketAddress(new InetSocketAddress(otherServiceInfo.getAddress(), otherServiceInfo.getPort()));
            // 超时请求其他服务端
            response = failOverInvokeRequest(properties, discoveryService, serviceName, metadata);
        }
        return response;
    }

}
